package com.flink.common;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.AggregateOperator;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.FlatMapOperator;
import org.apache.flink.api.java.operators.UnsortedGrouping;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

/**
* flink批处理
 */
public class FlinkBatch {
    public static void main(String[] args) throws Exception {
        //创建执行环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        //读取文件,参数是wc.txt文件的路径
        DataSource<String> line_DS = env.readTextFile("wc.txt");

        /**
         * 使用flatMap算子进行处理
         * FlatMapFunction<String, Tuple2<String, Integer>>
         * 第一个参数String表示进来的数据类型是String类型的
         * 第二个参数Tuple2<String, Integer>表示出去的类型是二元组类型的，二元组中第一个参数是String类型第二个参数是Integer类型
         * 在读取到数据之后需要转换成元祖然后再进行统计,每个单词放入元祖中都放一个1
         * 例如(hello,1) (java,1) (hello,1) (flink,1)
         * 然后分组统计后面的数字就知道每个单子出现了多少次
         */
        FlatMapOperator<String, Tuple2<String, Integer>> word_tuple2 = line_DS.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            /**
             * 重写flatMap方法
             * flatMap(String value, Collector<Tuple2<String, Integer>> out)
             * String value表示进来的参数value是String类型的
             * Collector<Tuple2<String, Integer>> out Collector是收集器，出去的参数out是二元组类型
             * @param value
             * @param out
             * @throws Exception
             */
            @Override
            public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
                //使用split按照空格进行拆分
                String[] words = value.split(" ");
                //拆分的结果是一个数组，然后遍历拆分的结果，也就是遍历出来每一个单词
                for (String word : words) {
                    //将单词添加到二元组中，后面写1，就到达了(hello,1) (java,1) (hello,1) (flink,1)这样的效果
                    Tuple2<String, Integer> tuple2_of = Tuple2.of(word, 1);
                    //向下游发送数据，collect方法就是把数据发送到下游
                    out.collect(tuple2_of);
                }
            }
        });

        /**
         * 分组操作
         * 根据单词进行分组，这里是根据二元组中的索引来分组的
         * 例如(hello,1)那么第一个元素也就是hello它的索引下标是0，第二个元素的索引下标是1
         * 所以这里我们通过0 来根据第一个元素进行分组，相同单词会被分到一组
         */
        UnsortedGrouping<Tuple2<String, Integer>> word_by = word_tuple2.groupBy(0);
        //对单词分完组之后，需要聚合一下第二个参数，来统计数量，第二个参数的下标是1，那么就写1
        AggregateOperator<Tuple2<String, Integer>> word_sum = word_by.sum(1);
        //输出结果，这里需要抛异常
        word_sum.print();
    }
}